package com.ctrip.framework.apollo.configservice.controller;

import com.ctrip.framework.apollo.biz.config.BizConfig;
import com.ctrip.framework.apollo.biz.entity.ReleaseMessage;
import com.ctrip.framework.apollo.biz.message.ReleaseMessageListener;
import com.ctrip.framework.apollo.biz.message.Topics;
import com.ctrip.framework.apollo.biz.utils.EntityManagerUtil;
import com.ctrip.framework.apollo.common.exception.BadRequestException;
import com.ctrip.framework.apollo.configservice.service.ReleaseMessageServiceWithCache;
import com.ctrip.framework.apollo.configservice.util.NamespaceUtil;
import com.ctrip.framework.apollo.configservice.util.WatchKeysUtil;
import com.ctrip.framework.apollo.configservice.wrapper.DeferredResultWrapper;
import com.ctrip.framework.apollo.core.ConfigConsts;
import com.ctrip.framework.apollo.core.dto.ApolloConfigNotification;
import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;
import com.ctrip.framework.apollo.tracer.Tracer;
import com.google.common.base.Splitter;
import com.google.common.base.Strings;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;

import java.lang.reflect.Type;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * @author Jason Song(song_s@ctrip.com)
 */
@RestController
@RequestMapping("/notifications/v2")
public class NotificationControllerV2 implements ReleaseMessageListener {
  private static final Logger logger = LoggerFactory.getLogger(NotificationControllerV2.class);
  /**
   * Watch Key 与 DeferredResultWrapper 的 Multimap
   * Key：Watch Key
   * Value：DeferredResultWrapper 数组
   */
  private final Multimap<String, DeferredResultWrapper> deferredResults =
      Multimaps.synchronizedSetMultimap(HashMultimap.create());
  private static final Splitter STRING_SPLITTER =
      Splitter.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR).omitEmptyStrings();
  private static final Type notificationsTypeReference =
      new TypeToken<List<ApolloConfigNotification>>() {
      }.getType();
  /**
   * 大量通知分批执行 ExecutorService
   */
  private final ExecutorService largeNotificationBatchExecutorService;

  private final WatchKeysUtil watchKeysUtil;
  private final ReleaseMessageServiceWithCache releaseMessageService;
  private final EntityManagerUtil entityManagerUtil;
  private final NamespaceUtil namespaceUtil;
  private final Gson gson;
  private final BizConfig bizConfig;

  @Autowired
  public NotificationControllerV2(
      final WatchKeysUtil watchKeysUtil,
      final ReleaseMessageServiceWithCache releaseMessageService,
      final EntityManagerUtil entityManagerUtil,
      final NamespaceUtil namespaceUtil,
      final Gson gson,
      final BizConfig bizConfig) {
    largeNotificationBatchExecutorService = Executors.newSingleThreadExecutor(ApolloThreadFactory.create
        ("NotificationControllerV2", true));
    this.watchKeysUtil = watchKeysUtil;
    this.releaseMessageService = releaseMessageService;
    this.entityManagerUtil = entityManagerUtil;
    this.namespaceUtil = namespaceUtil;
    this.gson = gson;
    this.bizConfig = bizConfig;
  }

  /**
   * 客户端获取配置通知的接口
   * @param appId
   * @param cluster
   * @param notificationsAsString
   * @param dataCenter
   * @param clientIp   请求参数，目前该接口暂时用不到，作为预留参数。🙂 万一未来在灰度发布需要呢。
   * @return
   */
  @GetMapping
  public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification(
      @RequestParam(value = "appId") String appId,
      @RequestParam(value = "cluster") String cluster,
      @RequestParam(value = "notifications") String notificationsAsString,
      @RequestParam(value = "dataCenter", required = false) String dataCenter,
      @RequestParam(value = "ip", required = false) String clientIp) {
    List<ApolloConfigNotification> notifications = null;

    try {
      /* 客户端传过来的例子：kv对
       [{"namespaceName":"TEST1.apollo","notificationId":-1},
      {"namespaceName":"application","notificationId":-1}]
      */
      notifications =
          gson.fromJson(notificationsAsString, notificationsTypeReference);
    } catch (Throwable ex) {
      Tracer.logError(ex);
    }

    if (CollectionUtils.isEmpty(notifications)) {
      throw new BadRequestException("Invalid format of notifications: " + notificationsAsString);
    }
    // 将要返回到客户端的通知
    DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper();
    Set<String> namespaces = Sets.newHashSet(); // 该app对应的服务端的appNamespaceName
    // 创建客户端的通知信息 Map 。其中，KEY 为 Namespace 的名字，VALUE 为通知编号。
    Map<String, Long> clientSideNotifications = Maps.newHashMap();
    /***** 只取本次请求的appId对应的通知，key是服务端的appNamespaceName  **/
    Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications);

    for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) {
      String normalizedNamespace = notificationEntry.getKey();
      ApolloConfigNotification notification = notificationEntry.getValue();
      namespaces.add(normalizedNamespace);

      clientSideNotifications.put(normalizedNamespace, notification.getNotificationId());

      /** 如果客户端配置的NamespaceName和服务端不同，就记录下来，等返回给客户端时再转过来，否则客户端无法识别  ***/
      if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) {
        deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace);
      }
    }

    if (CollectionUtils.isEmpty(namespaces)) {
      throw new BadRequestException("Invalid format of notifications: " + notificationsAsString);
    }

    //获取客户端namespace对应的服务端watchKey,  key是namespace   value是appId+cluster+namespace
    Multimap<String, String> watchedKeysMap =
        watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter);

    //该app 所有对应的watchKey
    Set<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values());

    // 根据watchKey获取所有的最新发布信息
    List<ReleaseMessage> latestReleaseMessages =
        releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys);

    /**
     * Manually close the entity manager.
     * Since for async request, Spring won't do so until the request is finished,
     * which is unacceptable since we are doing long polling - means the db connection would be hold
     * for a very long time
     * ，手动关闭 EntityManager 。因为对于 async 请求，SpringMVC 在请求完成之前不会这样做。这是不可接受的，因为我们正在做长轮询——意味着 db 连接将被保留很长时间。实际上，下面的过程，我们已经不需要 db 连接，因此进行关闭。
     */
    //  entity manager jpa管理持久化的,可先不管它  https://blog.csdn.net/J080624/article/details/78751411
    entityManagerUtil.closeEntityManager();

    /**  获取每个namespace对应的最新发布    **/
    List<ApolloConfigNotification> newNotifications =
        getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap,
            latestReleaseMessages);

    if (!CollectionUtils.isEmpty(newNotifications)) {
      // 有新的通知,直接设置 DeferredResult 的结果，从而结束长轮询
      deferredResultWrapper.setResult(newNotifications);
    } else {  // 若无新的通知，注册到 deferredResults 中，等到有配置变更或超时。

      // 注册超时事件，超时后，记录1个日志
      deferredResultWrapper
          .onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));

      //   注册结束事件，意思是该http请求结束之后，执行onCompletion中的方法
      //   在其内部，移除注册的 Watch Key + DeferredResultWrapper 出 deferredResults
      deferredResultWrapper.onCompletion(() -> {
        // 下面的参数，都是this.deferredResults.put(key, deferredResultWrapper) 添加进来的
        //unregister all keys
        for (String key : watchedKeys) {
          // 意思是：将key对应的集合中，移除deferredResultWrapper

          // 说明：将onCompletion绑定到对象deferredResultWrapper上了，因此我们可以从deferredResults中，获取key对应的集合，然后移除绑定对象deferredResultWrapper
          deferredResults.remove(key, deferredResultWrapper);
        }
        logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");
      });

      // 注册 Watch Key + DeferredResultWrapper 到 deferredResults 中，等待配置发生变化后通知。这样，任意一个 Watch Key 对应的 Namespace 对应的配置发生变化时，都可以进行通知，并结束轮询等待。详细解析，见 「2.3 handleMessage」 方法。
      //register all keys
      //   此处添加的都是默认的DeferredResultWrapper对象
      for (String key : watchedKeys) {
        this.deferredResults.put(key, deferredResultWrapper);
      }

      logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys");
      logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}",
          watchedKeys, appId, cluster, namespaces, dataCenter);
    }

    return deferredResultWrapper.getResult();
  }

  /**
   * 只取本次请求的appId对应的通知，key是服务端的appNamespaceName
   *
   * @param appId
   * @param notifications
   * @return
   */
  private Map<String, ApolloConfigNotification> filterNotifications(String appId,
                                                                    List<ApolloConfigNotification> notifications) {
    Map<String, ApolloConfigNotification> filteredNotifications = Maps.newHashMap();
    for (ApolloConfigNotification notification : notifications) {
      if (Strings.isNullOrEmpty(notification.getNamespaceName())) {
        continue;
      }
      //strip out .properties suffix；  变量originalNamespace应该叫做filterNamespace，是处理后的名字，不是原始名
      String originalNamespace = namespaceUtil.filterNamespaceName(notification.getNamespaceName());
      notification.setNamespaceName(originalNamespace);
      //fix the character case issue, such as FX.apollo <-> fx.apollo
      // 归一化，将不同client的namespace都转为数据库中的namespace
      String normalizedNamespace = namespaceUtil.normalizeNamespace(appId, originalNamespace);

      //芋头解释： 如果客户端 Namespace 的名字有大小写的问题，并且恰好有不同的通知编号。例如 Namespace 名字为 "FX.apollo" 的通知编号是 1 ，但是 "fx.apollo" 的通知编号为 2 。我们应该让 "FX.apollo" 可以更新它的通知编号，所以，我们使用 "FX.apollo" 的 ApolloConfigNotification 对象，添加到结果，而忽略 "fx.apollo" 。通过这样的方式，若此时服务器的通知编号为 3 ，那么 "FX.apollo" 的通知编号先更新成 3 ，再下一次长轮询时，"fx.apollo" 的通知编号再更新成 3 。

      //官方解释： in case client side namespace name has character case issue and has difference notification ids
      // such as FX.apollo = 1 but fx.apollo = 2, we should let FX.apollo have the chance to update its notification id
      // which means we should record FX.apollo = 1 here and ignore fx.apollo = 2

      //我的解释： 如果filteredNotifications包含normalizedNamespace，说明客户端配置中有1个配置，被归一化为了normalizedNamespace；
      // 而且发现客户端还有1个配置，也被归一化为了normalizedNamespace，并且这次对应的通知id,比上个通知id大，就不处理它
      // 比如客户端同时配置FX.apollo = 1和fx.apollo = 2；那么这次就先更新FX.apollo = 1，下一轮再更新fx.apollo = 2
      if (filteredNotifications.containsKey(normalizedNamespace) &&
          filteredNotifications.get(normalizedNamespace).getNotificationId() < notification.getNotificationId()) {
        continue;
      }

      filteredNotifications.put(normalizedNamespace, notification);
    }
    return filteredNotifications;
  }

  /**
   * 获取每个namespace对应的最新发布
   * @param namespaces
   * @param clientSideNotifications
   * @param watchedKeysMap
   * @param latestReleaseMessages
   * @return
   */
  private List<ApolloConfigNotification> getApolloConfigNotifications(Set<String> namespaces,
                                                                      Map<String, Long> clientSideNotifications,
                                                                      Multimap<String, String> watchedKeysMap,
                                                                      List<ReleaseMessage> latestReleaseMessages) {
    List<ApolloConfigNotification> newNotifications = Lists.newArrayList();
    if (!CollectionUtils.isEmpty(latestReleaseMessages)) {
      // 创建最新通知的 Map 。其中 Key 为 Watch Key 。
      Map<String, Long> latestNotifications = Maps.newHashMap();
      for (ReleaseMessage releaseMessage : latestReleaseMessages) {
        latestNotifications.put(releaseMessage.getMessage(), releaseMessage.getId());
      }
      /** 查找每个namespace对应的最新发布 ***************/
      for (String namespace : namespaces) {
        long clientSideId = clientSideNotifications.get(namespace); // 客户端目前的配置id
        long latestId = ConfigConsts.NOTIFICATION_ID_PLACEHOLDER;  // 最新的配置id
        Collection<String> namespaceWatchedKeys = watchedKeysMap.get(namespace);
        /** 查看1个namespace对应的所有WatchKey,找到最大的配置id ************/
        for (String namespaceWatchedKey : namespaceWatchedKeys) {
          long namespaceNotificationId =
              latestNotifications.getOrDefault(namespaceWatchedKey, ConfigConsts.NOTIFICATION_ID_PLACEHOLDER);
          if (namespaceNotificationId > latestId) {
            latestId = namespaceNotificationId;
          }
        }
        /** 将该namespace对应的最新配置，存到newNotifications  **/
        if (latestId > clientSideId) {
          ApolloConfigNotification notification = new ApolloConfigNotification(namespace, latestId);
          namespaceWatchedKeys.stream().filter(latestNotifications::containsKey).forEach(namespaceWatchedKey ->
              notification.addMessage(namespaceWatchedKey, latestNotifications.get(namespaceWatchedKey)));
          newNotifications.add(notification);
        }
      }
    }
    return newNotifications;
  }

  //ApolloNotificationMessages  发布配置时触发：  当有新的 ReleaseMessage 时，通知其对应的 Namespace 的，并且正在等待的请求。

  @Override
  public void handleMessage(ReleaseMessage message, String channel) {
    logger.info("message received - channel: {}, message: {}", channel, message);
    // 例子：watchKey 100004458+default+application
    String content = message.getMessage();
    Tracer.logEvent("Apollo.LongPoll.Messages", content);
    // 仅处理 APOLLO_RELEASE_TOPIC
    if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(content)) {
      return;
    }

    String changedNamespace = retrieveNamespaceFromReleaseMessage.apply(content);

    if (Strings.isNullOrEmpty(changedNamespace)) {
      logger.error("message format invalid - {}", content);
      return;
    }

    // 客户端监听 deferredResults 中 存在对应的 Watch Key。
    if (!deferredResults.containsKey(content)) {
      return;
    }

    // 从 deferredResults 中读取并创建 DeferredResultWrapper 数组，避免并发问题。
    //create a new list to avoid ConcurrentModificationException
    List<DeferredResultWrapper> results = Lists.newArrayList(deferredResults.get(content));

    ApolloConfigNotification configNotification = new ApolloConfigNotification(changedNamespace, message.getId());
    // 添加通知消息明细。此处，details 是一个。
    configNotification.addMessage(content, message.getId());

    // 假设一个公共 Namespace 有10W 台机器使用，如果该公共 Namespace 发布时直接下发配置更新消息的话，就会导致这 10W 台机器一下子都来请求配置，这动静就有点大了，而且对 Config Service 的压力也会比较大。
    //do async notification if too many clients
    // 若需要通知的客户端过多，使用 ExecutorService 异步通知，避免“惊群效应”
    if (results.size() > bizConfig.releaseMessageNotificationBatch()) {
      largeNotificationBatchExecutorService.submit(() -> {
        logger.debug("Async notify {} clients for key {} with batch {}", results.size(), content,
            bizConfig.releaseMessageNotificationBatch());
        for (int i = 0; i < results.size(); i++) {
           // 每 N 个客户端，sleep 一段时间。
          if (i > 0 && i % bizConfig.releaseMessageNotificationBatch() == 0) {
            try {
              TimeUnit.MILLISECONDS.sleep(bizConfig.releaseMessageNotificationBatchIntervalInMilli());
            } catch (InterruptedException e) {
              //ignore
            }
          }
          logger.debug("Async notify {}", results.get(i));
          results.get(i).setResult(configNotification);
        }
      });
      return;
    }

    logger.debug("Notify {} clients for key {}", results.size(), content);

    // 设置 DeferredResult 的结果，从而结束长轮询
    for (DeferredResultWrapper result : results) {
      result.setResult(configNotification);
    }
    logger.debug("Notification completed");
  }
  /**
   * 通过 ReleaseMessage 的消息内容，获得对应 Namespace 的名字
   */
  private static final Function<String, String> retrieveNamespaceFromReleaseMessage =
      releaseMessage -> {
        if (Strings.isNullOrEmpty(releaseMessage)) {
          return null;
        }
        List<String> keys = STRING_SPLITTER.splitToList(releaseMessage);
        //message should be appId+cluster+namespace
        if (keys.size() != 3) {
          logger.error("message format invalid - {}", releaseMessage);
          return null;
        }
        return keys.get(2);
      };

  private void logWatchedKeys(Set<String> watchedKeys, String eventName) {
    for (String watchedKey : watchedKeys) {
      Tracer.logEvent(eventName, watchedKey);
    }
  }
}

